/**
 * 
 */
package com.etlmaster.datax.plugins.writer.vfswriter;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.net.URI;
import java.util.zip.GZIPOutputStream;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.vfs2.FileObject;
import org.apache.hadoop.io.compress.CompressionOutputStream;

import com.dp.nebula.wormhole.common.AbstractPlugin;
import com.dp.nebula.wormhole.common.interfaces.ILine;
import com.dp.nebula.wormhole.common.interfaces.ILineReceiver;
import com.dp.nebula.wormhole.common.interfaces.IWriter;
import com.dp.nebula.wormhole.plugins.common.SFTPUtils;
import com.etlmaster.datax.plugins.writer.sftpwriter.ParamKey;
import com.etlmaster.executor.utils.LogWriter;
import com.hadoop.compression.lzo.LzopCodec;

/**
 * @author FengL
 *
 */
public class VfsWriter extends AbstractPlugin implements IWriter{
 	
	private String dir = "";
	private String fileType = "txt";
	private String fieldSplit = "\t";
	private char lineSplit = '\n';
	private String encoding = "UTF-8";
	private int bufferSize = 4 * 1024;
	private char[] nullChars = null;
	
	
	
	private GZIPOutputStream gout = null;
	private CompressionOutputStream cout = null;
	private OutputStream out = null;
	private BufferedWriter bw = null;
	private FileObject targetFile = null;
	
	/**
	 * 
	 */
	public VfsWriter() {
		// TODO Auto-generated constructor stub
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

	}
	
	/**
	 * 
	 * @see 初始化参数
	 */
	@Override
	public void init() {		
		dir = getParam().getValue(ParamKey.dir, this.dir).trim();
		if (StringUtils.isBlank(dir)) {
			LogWriter.addLog("ERROR","Can't find the param [" + ParamKey.dir
					+ "] in sftp-spliter-param.");
			return;
		}
		fileType = getParam().getValue(ParamKey.fileType, this.fileType).trim();
		bufferSize = getParam().getIntValue(ParamKey.bufferSize,
				this.bufferSize);
		fieldSplit = getParam().getValue(ParamKey.fieldSplit,
				this.fieldSplit);
		lineSplit = getParam().getCharValue(ParamKey.lineSplit, this.lineSplit);
		encoding = getParam().getValue(ParamKey.encoding, this.encoding);
		nullChars = getParam().getValue(ParamKey.nullChar, "").toCharArray();
	}
	@Override
	public void connection(){
		
		try{
		//	VFSFileFinder finder = new VFSFileFinder();
			targetFile=VFSFileFinder.resolveFile(dir);
			out=targetFile.getContent().getOutputStream(false);
		if (fileType.equalsIgnoreCase("txt")) {
			bw = new BufferedWriter(new OutputStreamWriter(out, encoding),
					bufferSize);
		} else if (fileType.equalsIgnoreCase("gz")
				|| fileType.equalsIgnoreCase("gzip")) {
			gout = new GZIPOutputStream(out);
			bw = new BufferedWriter(new OutputStreamWriter(gout, encoding),
					bufferSize);
		} else if (fileType.equalsIgnoreCase("lzo")) {
			LzopCodec lzopCodec = new LzopCodec();
			lzopCodec.setConf(SFTPUtils.getConf());

			cout = lzopCodec.createOutputStream(out);
			bw = new BufferedWriter(new OutputStreamWriter(cout, encoding),
					bufferSize);
		} else {
			throw new IllegalArgumentException("illegal argument fileType="
					+ fileType);
		}
		}catch(Exception e){
			closeAll();
			throw new RuntimeException("something wrong with vfs:"
					+ e.getMessage());
		}
	}
	
	public void write(ILineReceiver lineReceiver) {
		ILine line = null;
		int count=0;
		try {
			while ((line = lineReceiver.receive()) != null) {
				int len = line.getFieldNum();
				for (int i = 0; i < len; i++) {
					bw.write(replaceChars(line.getField(i)));
					if (i < len - 1) {
						if(fieldSplit.equalsIgnoreCase("\\t")){
							bw.write('\t');
						}else
							bw.write(fieldSplit);
					}
				}
				bw.write(lineSplit);
				if(getMonitor()!=null) {
					getMonitor().increaseSuccessLines();
				}
				count++;
				if(count==100000)
				bw.flush();
			}
			
		} catch (Exception e) {
			LogWriter.addLog("ERROR",e.getMessage());
		}
		
	}
	private char[] replaceChars(String str) {
		if (null == str) {
			return this.nullChars;
		}
		str=str.replace("\n", " ");
		return str.toCharArray();
	}
	public void commit() {
		// TODO Auto-generated method stub
		
	}
	
	@Override
	public void finish() {
		closeAll();
	}
	
	public void closeAll(){
		try {
			if (bw != null) {
				bw.close();
			}
			if (cout != null) {
				cout.close();
			}
			if (gout != null) {
				gout.close();
			}
			if (out != null) {
				out.close();
			}
			if(targetFile!=null){
				targetFile.close();
			}
//			if (c != null) {
//				c.disconnect();
//			}
//			if (session != null) {
//				session.disconnect();
//			}
		} catch (IOException e) {
			LogWriter.addLog("ERROR",e.getMessage());
		}
	}

}
